package com.lvlian;

import com.ConfigLoader;
import com.DatabaseConfig;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;
import java.util.Properties;

public class LvlianSouce {
    public static MySqlSource<String> createCDCSource(String  fileName,StartupOptions startPos) {

        DatabaseConfig config = ConfigLoader.loadConfig( fileName);
        System.out.println(  "输出数据库信息  ========"+config.toString());
        Properties prop = new Properties();
        prop.setProperty("decimal.handling.mode", "string");

        prop.setProperty("debezium.snapshot.mode", "initial");
        prop.setProperty("snapshot.mode", "initial"); // 确保捕获所有数据


        return MySqlSource.<String>builder()
                .hostname(config.getSourceHostname())
                .port(config.getSourcePort())
                .username(config.getSourceUsername())
                .password(config.getSourcePassword())
                .databaseList(config.getSourceDatabase())
                .tableList(config.getSourceDatabase()+".scf_recycling_in_info" ,
                        config.getSourceDatabase()+".scf_renewable_in_info",
                        config.getSourceDatabase()+".scf_invoice_info",
                        config.getSourceDatabase()+".scf_payment_info"
                        ,config.getSourceDatabase()+".scf_contract_info",
                         config.getSourceDatabase()+".scf_driver_info",
                         config.getSourceDatabase()+".scf_truck_info",
                        config.getSourceDatabase()+".scf_subject_basic_info"
                )
                .startupOptions(startPos)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .debeziumProperties(prop)
                .build();


    }
}
